package studio.raptor.extractor.segment.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import studio.raptor.databus.command.Event;
import studio.raptor.databus.command.EventBasicData;
import studio.raptor.databus.command.EventMetaData;
import studio.raptor.extractor.common.constant.EventMetaStatus;
import studio.raptor.extractor.common.holder.ExecutorHolder;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.dao.EventDao;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.server.ExtractorServer;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by liuyong8 on 2017/7/18.
 */
@Service
public class EventMetaService {

  private static final Logger log = LoggerFactory.getLogger(ExtractorServer.class);

  @Autowired
  private EventDao eventDao;

  private ServerConfig serverConfig;

  private BlockingQueue<EventMeta> eventQueue;

  private final ExecutorService checkRecoverProgress =
      ExecutorHolder.createSingleThreadPool("check_recover_progress");


  public Long addEventMetaData(EventBasicData eventBasicData, EventMetaData eventMetaData) {
    return eventDao.addEventMeta(eventBasicData, eventMetaData);
  }

  public List<EventMeta> findEventMetas(Integer kindId) throws Exception {
    List<EventMeta> eventMetas =
        eventDao.queryEventMetaByKindIdAndStatus(kindId, EventMetaStatus.STATUS_WAIT.value());
    return eventMetas;
  }

  public void modifyEventMetaStatus2P(EventMeta eventMeta) throws SQLException {
    eventMeta.setStatus(EventMetaStatus.STATUS_PROCESS.value());
    eventDao.updateEventMetaStatusByKindIdAndObjectKey(eventMeta);
  }

  public void modifyEventMetaStatus2C(EventMeta eventMeta) throws SQLException{
    eventMeta.setStatus(EventMetaStatus.STATUS_COMPLETE.value());
    eventDao.updateEventMetaStatusByKindIdAndObjectKey(eventMeta);
  }

  public void modifyEventMetaStatus2F(EventMeta eventMeta) throws SQLException{
    eventMeta.setStatus(EventMetaStatus.STATUS_FALSE.value());
    eventDao.updateEventMetaStatusByKindIdAndObjectKey(eventMeta);
  }

  /**
   * 处理停止前未处理的eventMeta
   */
  public void recover() throws Exception {
    serverConfig.setExtractorMQEnabled(new AtomicBoolean(false));
    boolean shouldContinue = true;

    // step1:recover event meta P -> W
    List<EventMeta> eventMetas = eventDao.queryEventMetaByKindIdAndStatus(serverConfig.getKindId(),
        EventMetaStatus.STATUS_PROCESS.value());

    if (eventMetas.isEmpty()) {
      log.info("events in processing is empty，nothing to recover!");
    } else {
      log.info("begin try to reset ordered-event state:[P->W]", eventMetas.size());
      for (EventMeta eventMeta : eventMetas) {
        try {
          eventMeta.setStatus(EventMetaStatus.STATUS_WAIT.value());
          eventMeta.setRemark("recover");
          eventDao.updateEventMetaStatusByKindIdAndObjectKey(eventMeta);
          // step3:放入抽取队里
          log.info("begin put into queue");
          eventQueue.put(eventMeta);
        } catch (Exception e) {
          log.error("recover " + eventMeta.getObjectKey() + "fail ", e);
          shouldContinue = false;
          break;
        }
      }
      log.info("end try to reset ordered-event state:[P->W]");


    }

    if (shouldContinue) {
      checkRecoverProgress.submit(new Runnable() {
        @Override
        public void run() {
          // step2:event_meta表中剩余条数需要抽取
          List<String> statuses = new ArrayList<>();
          statuses.add(EventMetaStatus.STATUS_WAIT.value());
          statuses.add(EventMetaStatus.STATUS_PROCESS.value());
          while (1 == 1) {
            List<EventMeta> eventMetas =
                eventDao.queryMetasByStatuses(serverConfig.getKindId(), statuses);
            int leftHandleEventNum = eventMetas.size();
            if (leftHandleEventNum == 0) { // 停止
              serverConfig.setExtractorMQEnabled(new AtomicBoolean(true));
              System.out.println("todo's event > left:0");
              break;
            } else {
              System.out.println("todo's event > left:" + leftHandleEventNum);
            }
          }
        }
      });
    }



  }

  public ServerConfig getServerConfig() {
    return serverConfig;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }

  public BlockingQueue<EventMeta> getEventQueue() {
    return eventQueue;
  }

  public void setEventQueue(BlockingQueue<EventMeta> eventQueue) {
    this.eventQueue = eventQueue;
  }
}
